package edu.tyut.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SQLDemo {
    public static void main(String[] args) {
        // 1. 创建表环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 2. 创建表环境
        tEnv.executeSql("create table source(" +
                "id int," +
                "ts bigint," +
                "vc int" +
                ")with(" +
                "'connector' = 'datagen'," +
                "'rows-per-second' = '1'," +
                "'fields.id.kind' = 'random'," +
                "'fields.id.min' = '1'," +
                "'fields.id.max' = '10'," +
                "'fields.ts.kind' = 'sequence'," +
                "'fields.ts.start' = '1'," +
                "'fields.ts.end' = '1000000'," +
                "'fields.vc.kind' = 'random'," +
                "'fields.vc.min' = '1'," +
                "'fields.vc.max' = '100'" +
                ");");

        tEnv.executeSql("create table sink(" +
                "id int," +
                "ts bigint," +
                "vs int" +
                ")with(" +
                "'connector' = 'print' " +
                ");");

        Table sourceTable = tEnv.sqlQuery("select id, sum(vc) as sumVC from source where id > 5 group by id ");
        sourceTable.execute().print();
        // tEnv.createTemporaryView("tmp", sourceTable);

        // Table API
    }
}
